/*
 * Proto Streamer
 * Copyright (C) 2009 John Pritchard.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License as
 * published by the Free Software Foundation; either version 2 of the
 * License, or (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301 USA.
 */
package p.s.m;

import p.s.Request;
import p.s.Output;

import com.google.protobuf.CodedOutputStream;
import com.google.protobuf.Message;

import java.io.IOException;

/**
 * Application downlink.
 *
 * @see Session
 * @author jdp
 */
public abstract class Get
    extends Session
    implements Consumer
{
    private volatile int service;
    private volatile long version;
    private volatile int pointer;


    public Get(){
        super();
    }

    public final Object getConsumerIdentity(){
        return this.methodIdentifier;
    }
    public final int getServicePointer(int service, long version){
        if (service == this.service && version == this.version)
            return this.pointer;
        else
            return 0;
    }
    public final void setServicePointer(int service, long version, int pointer){
        this.service = service;
        this.version = version;
        this.pointer = pointer;
    }

    protected abstract void muxEnter(CodedOutputStream out)
        throws IOException;

    protected abstract void muxExit();

    protected abstract void muxWrite(Message m, CodedOutputStream out)
        throws IOException;

    protected abstract Message heartbeat();

    public final void tail(Request request, Output out)
        throws IOException
    {
        Service session = this.setService(request.getPath());
        CodedOutputStream cout = CodedOutputStream.newInstance(out);
        try {
            this.muxEnter(cout);
            int hb = 0;
            while (this.open){
                synchronized(this.sessionMonitor){
                    session = this.session;
                }
                Message m = session.update(this,200L);

                if (null != m){
                    this.muxWrite(m,cout);
                    hb = 0;
                }
                else {
                    hb += 1;
                    if (4 == hb){
                        m = this.heartbeat();
                        if (null != m)
                            this.muxWrite(m,cout);
                        hb = 0;
                    }
                    request.classLoaderMark();
                }
            }
        }
        finally {
            this.muxExit();
        }
    }
    protected final void enter(Service session){
        this.session = session;
        session.enterConsumer(this);
    }
    protected final void exit(Service session){
        this.session = null;
        session.exitConsumer(this);
    }

}
